package com.xiongjie;

import org.infinispan.Cache;
import org.infinispan.configuration.cache.CacheMode;
import org.infinispan.configuration.cache.Configuration;
import org.infinispan.configuration.cache.ConfigurationBuilder;
import org.infinispan.configuration.global.GlobalConfiguration;
import org.infinispan.configuration.global.GlobalConfigurationBuilder;
import org.infinispan.manager.DefaultCacheManager;
import org.infinispan.manager.EmbeddedCacheManager;
import org.infinispan.notifications.Listener;
import org.infinispan.notifications.cachelistener.annotation.CacheEntryCreated;
import org.infinispan.notifications.cachelistener.event.CacheEntryCreatedEvent;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Map;

/**
 * 集群缓存的使用。集群效果只能在命令行中看到，IDE中无法得到指定效果
 * 通过注解设置类为事件监听类
 */
@Listener
public class ClusterNode {

    private final boolean useXmlConfig;
    private final String cacheName;
    private final String nodeName;
    private volatile boolean stop = false;

    public ClusterNode(boolean useXmlConfig, String cacheName, String nodeName) {
        this.useXmlConfig = useXmlConfig;
        this.cacheName = cacheName;
        this.nodeName = nodeName;
    }

    /**
     * 1.使用infinispan.xml配置,复本模式
     */
    public static void clusterByInfinispanAndRepl(){
        new ClusterNode(true, "repl", "A").run();
        new ClusterNode(true, "repl", "B").run();
    }

    /**
     * 2.使用infinispan.xml配置,分布式模式
     */
    public static void clusterByInfinispanAndDist(){
        new ClusterNode(true, "dist", "C").run();
        new ClusterNode(true, "dist", "D").run();
        new ClusterNode(true, "dist", "E").run();
    }

    /**
     * 3.使用jgroup.xml配置，复本模式
     */
    public static void clusterByJgroupAndRepl() {
        new ClusterNode(false, "repl", "F").run();
        new ClusterNode(false, "repl", "G").run();
    }

    /**
     * 4.使用jgroup.xml配置，分布式模式
     */
    public static void clusterByJgroupAndDist() {
        new ClusterNode(false, "dist", "H").run();
        new ClusterNode(false, "dist", "I").run();
        new ClusterNode(false, "dist", "J").run();
    }

    /**
     * 5.通用的集群启动方式
     */
    public static void clusterByCommon(boolean useXmlConfig,String cacheName,String nodeName) {
        new ClusterNode(useXmlConfig, cacheName, nodeName).run();
    }

    /**
     * 6.打印缓存内容
     * @param cache
     */
    public void printCacheContents(Cache<String, String> cache) {
        System.out.println( String.format("在 %s 节点地址上的缓存内容如下:", cache.getAdvancedCache().getRpcManager().getAddress() ) );

        ArrayList<Map.Entry<String, String>> entries = new ArrayList<>(cache.entrySet());
        Collections.sort(entries, (o1,o2) ->
                o1.getKey().compareTo(o2.getKey())
        );
        for (Map.Entry<String, String> e : entries) {
            System.out.println( String.format("\tkey的值为：%s,value的值为：%s", e.getKey(), e.getValue()) );
        }
        System.out.println();
    }

    /**
     * 7.注解事件监听方法
     */
    @CacheEntryCreated
    public void observeAdd(CacheEntryCreatedEvent<String, String> event) {
        if (event.isPre())
            return;

        System.out.println( String.format("\t缓存实体 %s 添加到缓存 %s 内部", event.getKey(), event.getCache()) );
    }

    /**
     * 8.官方代码创建缓存管理器
     */
    public EmbeddedCacheManager createCacheManager(){
        EmbeddedCacheManager cacheManager = new DefaultCacheManager(
                GlobalConfigurationBuilder.defaultClusteredBuilder()
                        .transport().nodeName(nodeName).addProperty("configurationFile", "jgroups.xml")
                        .build(),
                new ConfigurationBuilder()
                        .clustering()
                        .cacheMode(CacheMode.REPL_SYNC)
                        .build()
        );

        cacheManager.defineConfiguration("dist", new ConfigurationBuilder()
                .clustering()
                .cacheMode(CacheMode.DIST_SYNC)
                .hash().numOwners(2)
                .build());

        return cacheManager;
    }

    /**
     * 9.自己创建缓存管理器
     */
    public EmbeddedCacheManager createCacheManagerBySelf(){
        GlobalConfiguration globalConfiguration=GlobalConfigurationBuilder.defaultClusteredBuilder()
                .transport().nodeName(nodeName).addProperty("configurationFile", "jgroups.xml")
                .build();
        EmbeddedCacheManager cacheManager = new DefaultCacheManager(globalConfiguration);

        Configuration configuration=null;
        if("repl".equals(cacheName)){
            configuration=new ConfigurationBuilder()
                    .clustering()
                    .cacheMode(CacheMode.REPL_SYNC)
                    .build();

        }else if("dist".equals(cacheName)){
            configuration=new ConfigurationBuilder()
                    .clustering()
                    .cacheMode(CacheMode.DIST_SYNC)
                    .hash().numOwners(2)
                    .build();
        }
        cacheManager.defineConfiguration(cacheName, configuration);

        return cacheManager;
    }
//mvn compile exec:java -Djava.net.preferIPv4Stack=true -Dexec.mainClass="com.xiongjie.ClusterNode" -Dexec.args="false dist A"
    public static void main(String[] args){
        boolean useXmlConfig = false;
        String cacheName = "repl";
        String nodeName = null;

        for (String arg : args) {
            switch (arg) {
                case "true":
                    useXmlConfig = true;
                    break;
                case "false":
                    useXmlConfig = false;
                    break;
                case "dist":
                    cacheName = "dist";
                    break;
                case "repl":
                    cacheName = "repl";
                    break;
                default:
                    nodeName = arg;
                    break;
            }
        }
        clusterByCommon(useXmlConfig, cacheName, nodeName);

//        clusterByInfinispanAndRepl();
//        clusterByInfinispanAndDist();
//        clusterByJgroupAndRepl();
//        clusterByJgroupAndDist();
    }

    /**
     * 主要的infinispan集群业务代码
     * @throws IOException
     * @throws InterruptedException
     */
    public void run() {
        System.out.println("开始获取缓存管理器......");
        EmbeddedCacheManager cacheManager = null;
        if (useXmlConfig) {
            System.out.println("使用xml配置来启动缓存管理器");
            System.setProperty("nodeName", nodeName);
            try {
                cacheManager=new DefaultCacheManager("infinispan.xml");
            } catch (IOException e) {
                e.printStackTrace();
            }

        } else {
            System.out.println("使用自定义代码配置来启动缓存管理器");

//            cacheManager=createCacheManager();
            cacheManager=createCacheManagerBySelf();
        }
        final Cache<String, String> cache = cacheManager.getCache(cacheName);


        System.out.println( String.format("缓存名字：%s 启动于地址：%s, 缓存成员是：%s", cacheName, cacheManager.getAddress(),
                cache.getAdvancedCache().getRpcManager().getMembers() ) );
        printCacheContents(cache);

        //添加事件监听器
        cache.addListener(new ClusterNode(false,"",""));

        System.out.println("开始启动子线程......");
        Thread cacheThread = new Thread() {
            @Override
            public void run() {
                int counter = 0;
                while (!stop) {
                    try {
                        cache.put("key-" + counter, "" + cache.getAdvancedCache().getRpcManager().getAddress() + "-" + counter);
                    } catch (Exception e) {
                        System.out.println("插入key值到缓存报错。key值=key-"+counter);
                    }
                    counter++;

                    //子线程睡眠时间长度，决定缓存的添加速度
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        break;
                    }
                }
            }
        };
        cacheThread.start();

        System.out.println("下面的代码功能是：点击回车键打印缓存内容, 点击Ctrl+D/Ctrl+Z停止线程");
        while (true) {
            try {
                if (!(System.in.read() > 0)) break;
            } catch (IOException e) {
                e.printStackTrace();
            }
            printCacheContents(cache);
        }

        stop = true;
        try {
            cacheThread.join(); //主线程等待子线程结束
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        cacheManager.stop();
        System.exit(0);
    }
}
